package com.lgCore.consumer.service.impl;

import com.alibaba.fastjson.JSON;
import com.lgCore.consumer.model.ClientZkData;
import com.lgCore.consumer.service.IZookServiceClient;
import com.lgCore.factory.RpcServiceZkFactory;
import com.lgCore.model.ServiceInfaceData;
import com.lgCore.model.ServiceInfaceDetailData;
import com.lgCore.model.ZKBaseData;
import com.lgCore.model.ZKNodeType;
import com.lgCore.provide.model.ServiceZkData;
import com.lgCore.util.ZkUtil;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.framework.listen.ListenerContainer;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_ADDED;

public class ZookClientImpl implements IZookServiceClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(ZookClientImpl.class);
    private String zkRootPath;

    private CuratorFramework client;

    private String rpcSystemRegion;

    public ZookClientImpl() {
    }

    public void initConfig(String zkRootPath, CuratorFramework client) {
        this.zkRootPath = zkRootPath;
        this.client = client;
    }

    /**
     * 初始化server zk信息
     *
     * @return
     */
    @Override
    public void initServer() throws Exception {
        String regionPath = "/" + zkRootPath + "/" + RpcServiceZkFactory.getProfilesActive() + "/" + RpcServiceZkFactory.getSystemRegionType();
        //获取全部节点信息
        Map<String, String> allNodes = ZkUtil.getAllNodes(client, regionPath);
        if (allNodes != null && allNodes.size() > 0) {
            for (String key : allNodes.keySet()) {
                Object object = allNodes.get(key);
                if (object != null) {
                    try {
                        ZKBaseData zkBaseData = JSON.parseObject(object.toString(), ZKBaseData.class);
                        if (zkBaseData.getZkNodeType().getType().equals(ZKNodeType.INTERFACE.getType())) {
                            ServiceInfaceData serviceInfaceData = JSON.parseObject(object.toString(), ServiceInfaceData.class);
                            Map<String, String> childAllNodes = ZkUtil.getAllNodes(client, regionPath + "/" + serviceInfaceData.getServicePath());
                            if (childAllNodes != null && childAllNodes.size() > 0) {
                                for (String childKey : childAllNodes.keySet()) {
                                    Object childObject = childAllNodes.get(childKey);
                                    addServiceData(childObject, childKey);
                                }
                            }
                        } else if (zkBaseData.getZkNodeType().getType().equals(ZKNodeType.INTERFACE_DETAIL.getType())) {
                            addServiceData(object, key);
                        }
                    } catch (Exception e) {
                        LOGGER.info("解析失败");
                    }
                }
            }
        }
    }

    public void addServiceData(Object object, String childKey) {
        ServiceInfaceDetailData serviceInfaceDetailData = JSON.parseObject(object.toString(), ServiceInfaceDetailData.class);
        String data = serviceInfaceDetailData.getIp() + ":" + serviceInfaceDetailData.getNettyPort();
        //zk地址唯一标识
        String zkIdea = childKey.substring(childKey.lastIndexOf(":") + 1);
        data = data + ":" + zkIdea;
        ClientZkData.putData(serviceInfaceDetailData.getServicePath(), data);
    }

    /**
     * 监控zk节点变化
     *
     * @return
     */
    @Override
    public void monitorZkNode() throws Exception {
//        ExecutorService executor = Executors.newCachedThreadPool();
//        executor.submit(() -> {
        String regionPath = "/" + zkRootPath + "/" + RpcServiceZkFactory.getProfilesActive() + "/" + RpcServiceZkFactory.getSystemRegionType();
        //监控
        TreeCache cache = new TreeCache(client, regionPath);
        TreeCacheListener listener = (client1, event) -> {
            System.out.println("事件类型：" + event.getType() +
                    " | 路径：" + (null != event.getData() ? event.getData().getPath() : null));
            try {
                if (event.getData() != null) {
                    ChildData data = event.getData();
                    String dataPath;
                    ServiceInfaceData serviceInfaceData = new ServiceInfaceData();
                    switch (event.getType()) {
                        case NODE_ADDED:
                            dataPath = new String(data.getData(), "UTF-8");
                            serviceInfaceData.setServicePath(dataPath);
                            serviceInfaceData.setZkServicePath(data.getPath());
                            ClientZkData.syncZkData("add", JSON.toJSONString(serviceInfaceData));
                            break;
                        case NODE_UPDATED:
                            dataPath = new String(data.getData(), "UTF-8");
                            serviceInfaceData.setServicePath(dataPath);
                            serviceInfaceData.setZkServicePath(data.getPath());
                            ClientZkData.syncZkData("edit", JSON.toJSONString(serviceInfaceData));
                            break;
                        case NODE_REMOVED:
                            dataPath = new String(data.getData(), "UTF-8");
                            serviceInfaceData.setServicePath(dataPath);
                            serviceInfaceData.setZkServicePath(data.getPath());
                            ClientZkData.syncZkData("del", JSON.toJSONString(serviceInfaceData));
                            break;
                        default:
                            break;
                    }
                }

            } catch (Exception e) {
//                e.printStackTrace();
                LOGGER.info("非法字符");
            }
        };
        cache.getListenable().addListener(listener);
        try {
            cache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
//        });


    }


}
